package com.atbeijing.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming12_Out {

    def main(args: Array[String]): Unit = {

        // TODO 创建环境对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")

        // 构建StreamingContext对象时，第二个参数表示数据 采集周期，以毫秒单位，一般以秒为单位使用
        val ssc = new StreamingContext(sparkConf, Seconds(3))

        ssc.checkpoint("cp")
        // 从socket中获取的数据是一行一行的字符串
        val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

        val wordsDS = socketDS.flatMap(_.split(" "))

        val wordToOneDS = wordsDS.map((_,1))

        // 窗口的大小必须为采集周期的整数倍
        // 窗口滑动时大小也必须为采集周期的整数倍
        //val windowDS = wordToOneDS.window(Seconds(9),Seconds(3))
        //val wordToCountDS = windowDS.reduceByKey(_+_)

        List(1,2,3,4)

        val windowDS = wordToOneDS.reduceByKeyAndWindow(
            (x, y) => {
                println(x + "+" + y)
                x + y
            },
            (x, y) => {
                println(x + "-" + y)
                x - y
            },
            Seconds(9),
            Seconds(3)
        )

        windowDS.foreachRDD(
            rdd => {
                //rdd.toDF.write.jdbc()
                // Driver => Connection
                // 所有的连接对象都是不能序列化
//                val conn = Connection
//                // Executor => jdbc
//                rdd.foreachPartition(
//                    // connection
//                )

            }
        )

        // 启动采集器
        ssc.start()
        // Driver需要等待采集器的结束
        // awaitTermination阻塞主线程的运行
        ssc.awaitTermination()

        // SparkStreaming不能在Driver程序中执行stop方法
        //ssc.stop()
    }
}
